package yunjiao.javatutorials.guava.concurrent.transform;

import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;

import java.util.Base64;
import java.util.concurrent.*;

/**
 * 对异步计算结果进行转换处理
 *
 * @author yangyunjiao
 */
public class DemoExample {

    // 多个异步操作的流水线处理
    public static void async() throws ExecutionException, InterruptedException {
        ListeningExecutorService executor =
                MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(5));

        String input = "input";
        // 第一步：处理输入
        ListenableFuture<String> step1 = Futures.transform(
                executor.submit(() -> input),
                s -> {
                    String s1 = s.toUpperCase();
                    System.out.println("Step1:" + s1);
                    return s1;
                },
                executor
        );

        // 第二步：添加前缀
        ListenableFuture<String> step2 = Futures.transform(
                step1,
                s -> {
                    String s2 = "PREFIX_" + s;
                    System.out.println("Step2:" + s2);
                    return s2;
                },
                executor
        );

        // 第三步：加密（模拟）
        ListenableFuture<String> step3 = Futures.transform(
                step2,
                s -> {
                    String s3 = Base64.getEncoder().encodeToString(s.getBytes());
                    System.out.println("Step3:" + s3);
                    return s3;
                },
                executor
        );

        System.out.println("最终结果：" + step3.get());
        executor.shutdown();
    }

    // 处理异常，并提供默认值或替代结果
    public static void fault() throws ExecutionException, InterruptedException {
        ListeningExecutorService executor =
                MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(5));

        // 模拟可能失败的网络请求
        ListenableFuture<String> networkFuture = executor.submit(() -> {
            if (Math.random() < 0.5) {
                throw new RuntimeException("Network error");
            }
            return "网络请求结果";
        });

        // 转换，处理异常情况
        ListenableFuture<String> networkTransform = Futures.transform(
                networkFuture,
                data -> data, // 成功则直接返回
                executor
        );

        ListenableFuture<String> result = Futures.catching(
                networkFuture,
                Throwable.class,
                throwable -> "发生异常，返回默认值",
                executor
        );

        System.out.println("结果：" + result.get());
        executor.shutdown();
    }

    // 基础使用
    public static void base() throws ExecutionException, InterruptedException {
        ListeningExecutorService executor =
                MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(5));

        // 原始 Future
        ListenableFuture<String> originalFuture = executor.submit(() -> "123");

        // 转换 Future：String -> Integer
        ListenableFuture<Integer> transformedFuture = Futures.transform(
                originalFuture,
                input -> Integer.parseInt(input) * 10,  // 转换函数
                MoreExecutors.directExecutor()     // 执行器
        );

        // 获取转换后的结果
        Integer result = transformedFuture.get(); // 返回 123
        System.out.println("结果：" + result);

        executor.shutdown();
    }

    // 超时处理
    public static void timeout() throws InterruptedException, ExecutionException, TimeoutException {
        ListeningExecutorService executor =
                MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(5));

        ListenableFuture<String> future = executor.submit(() -> {
            // 模拟长时间运行的任务
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return "任务完成";
        });


        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
        ListenableFuture<String> withTimeout = Futures.withTimeout(
                future,
                1, TimeUnit.SECONDS,
                scheduler
        );

        // 转换，处理超时情况（超时会在Future上抛出TimeoutException）
        ListenableFuture<String> result = Futures.transform(
                withTimeout,
                data -> data,
                executor
        );

        try {
            System.out.println("结果" + result.get(1, TimeUnit.SECONDS));
        } catch (TimeoutException e) {
            System.out.println("超时异常");
        }

        executor.shutdown();
    }
}
